package clusteringTest;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;




import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.ObjectWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;

import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Reducer.Context;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;

import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.mahout.math.SequentialAccessSparseVector;
import org.apache.mahout.math.Vector;
import org.apache.mahout.math.VectorWritable;

import test.ReaderFromHDFS;

public class HbaseToVectorMapReduce {

/**
  * @author dxl
  * attention: this class have some problems,donot use it.just use HbaseToVector.java is OK.
  */
 /**
  * @fuction read data from hbase and transform them to vectors 
  * @throws Exception
  */
	
  public static class VectorMapper extends TableMapper <Text, Text> 
  {  
//    private final static IntWritable one = new IntWritable(1);
      
    public void map(ImmutableBytesWritable row, Result values,Context context) throws IOException, InterruptedException
    { 	      
	     Text ID = new Text();
	     Text CONTENT = new Text();
	     
	     for (KeyValue kv : values.list()) {
	       if ("html".equals(Bytes.toString(kv.getFamily())))
	       {
		    	String id = Bytes.toString(kv.getValue());
		    	ID.set(id);	   
		    	System.out.println("id: "+ id);
		    }//if
	       if ("title".equals(Bytes.toString(kv.getFamily()))) 
	       {
	    	   String content = Bytes.toString(kv.getValue());
	    	   CONTENT.set(content);
	    	   System.out.println("content: "+ content);
	       }//if  

      }//for

	      context.write(ID, CONTENT);
	     System.out.println("map end");
    }//map
  }
  
  public static class VectorSumReducer extends Reducer<Text,Text,Text,Text>
  {
	private Text result = new Text();
	public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException
	{
	     
		 String text ="";
		 for (Text val : values) 
		 { text += val.toString(); }//for
		 System.out.println("text  "+text);
		 
		 VectorizingText VT =new VectorizingText();
	     SequentialAccessSparseVector point = VT.TransformToVector(text);// name text  
		 System.out.println("Vector  " + point.toString());
		 
		 result.set(point.toString());
		 
		 Configuration conf = new Configuration(true);
		 FileSystem fs = FileSystem.get(conf);
		 Path path = new Path("testdata/points/file1");
		 SequenceFile.Writer writer = null;//new SequenceFile.(fs, conf, path, Text.class, VectorWritable.class);
		 writer = SequenceFile.createWriter(fs, conf, path,Text.class, VectorWritable.class);	
		 VectorWritable vec = new VectorWritable();
		 vec.set(point);
		 //recNum作为key，vector作为value
		 writer.append(key, vec);
		 writer.close();
		  
		 context.write(key, result);
		 System.out.println("combine/reduce  end");
	}//reduseblog_page
  }
  public static void main(String[] args) throws Exception {  
    Configuration conf = new Configuration();
    conf.set("hadoop.job.ugi", "tuxinhui,tuxinhui");
    conf = HBaseConfiguration.create(conf);
    
    
    Job job = new Job(conf, "clustering test");
    job.setJarByClass(HbaseToVectorMapReduce.class);
    
    job.setReducerClass(VectorSumReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(Text.class);
    
    Scan scan = new Scan();
    scan.setCaching(500); 
    scan.setCacheBlocks(false); 
    List<Filter> filters = new ArrayList<Filter>();
    
    Filter filter1 = new SingleColumnValueFilter(Bytes.toBytes("html"), null, CompareOp.GREATER_OR_EQUAL, Bytes.toBytes("bbb0")); // 当列column1的值为aaa时进行查询
    filters.add(filter1);
    Filter filter2 = new SingleColumnValueFilter(Bytes.toBytes("html"), null, CompareOp.LESS_OR_EQUAL, Bytes.toBytes("bbb2")); // 当列column1的值为aaa时进行查询
    filters.add(filter2);
    
    FilterList filterList = new FilterList(filters);
    scan.setFilter(filterList);
//  scan.addFamily(Bytes.toBytes("title"));//read data of title
    TableMapReduceUtil.initTableMapperJob("blog_page", scan,HbaseToVectorMapReduce.VectorMapper.class, Text.class, Text.class, job);   
    FileOutputFormat.setOutputPath(job, new Path("hdfs://localhost:9000/user/tuxinhui/output"));
    job.waitForCompletion(true);
    
//    ReaderFromHDFS RF = new ReaderFromHDFS();
//	String result = RF.ReaderHDFS("hdfs://localhost:9000/user/tuxinhui/output");//part-r-00000
//	System.out.println("The final result:");
//	System.out.println(result);

  }
}